package com.chenwc.rocketmq.example;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.UUID;

/**
 * 顺序消息发送
 * 顺序消息指的是按照消息发送的顺序，确保消息在消费者端也能按照相同的顺序进行消费。这种特性在需要严格顺序处理的场景中非常重要，例如订单处理、金融交易等。
 * RocketMQ实现顺序消息的原理主要依赖于消息队列的分区机制。生产者在发送消息时，通过特定的策略（如消息的key）选择发送到相应的队列。
 * 消费者从特定的队列中按顺序拉取消息，从而保证了消息的消费顺序与生产顺序一致。
 * 通过这种方式，RocketMQ可以保证单个队列中的消息顺序，但无法保证跨队列的全局顺序。
 *
 * @author chenwc
 */
public class ProducerFifoMessageExample {
    private static final Logger logger = LoggerFactory.getLogger(ProducerFifoMessageExample.class);

    private ProducerFifoMessageExample() {
    }

    public static void main(String[] args) throws MQClientException {

        String topic = "TopicTest";
        final DefaultMQProducer producer = ProducerSingleton.getInstance(topic);

        try {
            // 设置队列选择器
            // RocketMQ可以保证单个队列中的消息顺序，但无法保证跨队列的全局顺序，这里我们设置全部选择 Queue 0
            MessageQueueSelector mqs = new MessageQueueSelector() {
                /**
                 * @param mqs 当前Topic的队列列表
                 * @param msg 消息对象
                 * @param arg 消息选择器参数，这个也是 producer.send 方法传入的第三个参数
                 * @return 返回一个队列，即当前消息发送到这个队列，当所有消息都在同一个队列中，则保证顺序性
                 */
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // 如果当前Topic的队列列表不为空，则选择 Queue 0
                    if (mqs != null && !mqs.isEmpty()) {
                        return mqs.get(0);
                    }
                    return null;
                }
            };

            for (int i = 0; i < 10; i++) {
                // 定义消息体
                byte[] body = ("This is a fifo message for Apache RocketMQ, Message Index: " + (i + 1)).getBytes(StandardCharsets.UTF_8);
                // 定义消息标签
                String tag = "TagA";
                // 定义消息key
                String keys = UUID.randomUUID().toString().toUpperCase().replace("-", "");
                // 创建消息对象
                Message message = new Message(topic, tag, keys, body);

                // 发送消息
                // 这里的 i 是消息选择器参数，选择器会用到
                SendResult sendResult = producer.send(message, mqs, i);
                logger.info("Send message successfully, {}", sendResult.toString());
            }

        } catch (Throwable t) {
            logger.error("Failed to send message", t);
        }
        // 关闭生产者
        ProducerSingleton.closeInstance();
    }
}
